<!DOCTYPE html>
<html lang="zh-cn">
<head>
    <meta charset="utf-8" />
    
        <title>hadoopRPC | X-Blog</title>
    

    <meta http-equiv="X-UA-Compatible" content="IE=Edge,chrome=1"/>
<meta name="viewport" content="width=device-width,initial-scale=1.0,maximum-scale=1.0,user-scalable=no"/>
<meta name="format-detection" content="telephone=no">
<meta name="author" content="[object Object]" />
<meta name="designer" content="minfive" />
<meta name="keywords" content="Xiang blog, 程序员, SCI论文, Python, Hadoop, 深度炼丹"/>
<meta name="description" content="日常学习与兴趣交流的个人博客" />

<meta name="apple-mobile-web-app-capable" content="yes" />
<meta name="apple-mobile-web-app-status-bar-style" content="black" />
<meta name="format-detection" content="telephone=yes" />
<meta name="mobile-web-app-capable" content="yes" />
<meta name="robots" content="all" />

<link rel="canonical" href="https://spaces-x.github.io/2019/05/10/hadoopRPC/index.html">

<link rel="icon" type="image/png" href="https://www.easyicon.net/api/resizeApi.php?id=546977&size=32" sizes="32x32">














<!-- Prefetch -->






<!-- CSS -->
<link rel="stylesheet" href="/scss/base/index.css">

<!-- RSS -->
<link rel="alternate" href="/atom.xml" title="Space-X">

<!-- 统计 -->
<!-- 百度统计 -->


<!-- Global site tag (gtag.js) - Google Analytics -->


    
    
    <link rel="stylesheet" href="/scss/views/page/post.css">

</head>
<body ontouchstart>
    
        <!-- loading 页面 -->
<div id="page-loading" class="page page-loading" style="background-image: url('http://oo12ugek5.bkt.clouddn.com/blog/images/loader.gif')"></div>
    

    <div id="page" class="page js-hidden">
        
    <!-- 页头 -->
<header class="page__small-header page__header--small">
    <nav class="page__navbar">
        <div class="page__container navbar-container">
            <a class="page__logo" href="/" title="Space-X" alt="Space-X">
                <img src="https://s1.ax1x.com/2018/07/15/PMfvfP.png" alt="Space-X">
            </a>

            <nav class="page__nav">
                <ul class="nav__list clearfix">
                    
                        
                        <li class="nav__item">
                            <a href="/" alt="首页" title="首页">首页</a>
                        </li>
                    
                        
                        <li class="nav__item">
                            <a href="/archives" alt="归档" title="归档">归档</a>
                        </li>
                    
                        
                        <li class="nav__item">
                            <a href="/about" alt="关于" title="关于">关于</a>
                        </li>
                    
                </ul>
            </nav>

            <button class="page__menu-btn" type="button">
                <i class="iconfont icon-menu"></i>
            </button>
        </div>
    </nav>
</header>


        
    <main class="page__container page__main">
    <div class="page__content">
        <article class="page__post">
    <div class="post__cover">
        <img src="https://s2.ax1x.com/2019/05/10/EWAoB6.md.jpg" alt="hadoopRPC">
    </div>

    <header class="post__info">
        <h1 class="post__title">hadoopRPC</h1>

        <div class="post__mark">
            <div class="mark__block">
                <i class="mark__icon iconfont icon-write"></i>
                <ul class="mark__list clearfix">
                    <li class="mark__item">
                        <a href="https://www.github.com/spaces-x">WeiXiang</a>
                    </li>
                </ul>
            </div>
            
            <div class="mark__block">
                <i class="mark__icon iconfont icon-time"></i>
                <ul class="mark__list clearfix">
                    <li class="mark__item"><span>2019-05-10</span></li>
                </ul>
            </div>

            <div class="mark__block">
                <i class="mark__icon iconfont icon-tab"></i>
                <ul class="mark__list clearfix">
                    
                        <li class="mark__item">
                            <a href="/tags/Hadoop/">Hadoop</a>
                        </li>
                    
                        <li class="mark__item">
                            <a href="/tags/RPC/">RPC</a>
                        </li>
                    
                </ul>
            </div>

            
        </div>
    </header>

    <div class="post__content">
        <h1 id="Hadoop-RPC模块源码分析"><a href="#Hadoop-RPC模块源码分析" class="headerlink" title="Hadoop RPC模块源码分析"></a>Hadoop RPC模块源码分析</h1><h2 id="RPC概述"><a href="#RPC概述" class="headerlink" title="RPC概述"></a>RPC概述</h2><p>参考文章</p>
<blockquote>
<p><a href="https://www.cnblogs.com/qq503665965/p/6708644.html" target="_blank" rel="noopener">https://www.cnblogs.com/qq503665965/p/6708644.html</a></p>
</blockquote>
<h2 id="RPC组成"><a href="#RPC组成" class="headerlink" title="RPC组成"></a>RPC组成</h2><p>Hadoop RPC主要由三大类组成，即RPC、Client、和Server ，分别对应对外编程接口、客户端实现和服务器端实现。Hadoop 关于rpc的代码在hadoop-common下的org.apache.hadoop.ipc包中。</p>
<h2 id="类结构关系详解"><a href="#类结构关系详解" class="headerlink" title="类结构关系详解"></a>类结构关系详解</h2><p>类图是老版本的，部分函数名有变化但是架构没变。</p>
<ol>
<li><p><strong>ipc.RPC</strong></p>
<p>关键类图如下：</p>
<p><a href="https://imgchr.com/i/E8xkOx" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/04/30/E8xkOx.md.png" alt="E8xkOx.md.png"></a> </p>
</li>
<li><p><strong>ipc.Client</strong></p>
<p>关键类图分析如下：</p>
<p><a href="https://imgchr.com/i/E8x0cn" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/04/30/E8x0cn.md.png" alt="E8x0cn.md.png"></a></p>
</li>
</ol>
<ol start="3">
<li><p><strong>ipc.Server</strong></p>
<p><a href="https://imgchr.com/i/E8xBXq" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/04/30/E8xBXq.md.png" alt="E8xBXq.md.png"></a></p>
</li>
</ol>
<h2 id="源码分析"><a href="#源码分析" class="headerlink" title="源码分析"></a>源码分析</h2><h3 id="client-实现"><a href="#client-实现" class="headerlink" title="client 实现"></a>client 实现</h3><p>Client端实现结构如下图所示，从图中可以看出Client 包含两个内部类 Call和Connection</p>
<p><img src="https://s2.ax1x.com/2019/04/30/EGkhgU.png" alt="EGkhgU.png"></p>
<ol>
<li><p><strong>static class Call 内部类</strong></p>
<p>该类封装了一个RPC请求，它包含五个成员变量，分别是唯一标识<strong>id</strong>，函数调用信息<strong>rpcRequest</strong>、函数执行返回值<strong>rpcResponse</strong>，异常信息<strong>error</strong>和执行完成标识<strong>done</strong>。由于HadoopRPCServer采用了异步方式处理客户端请求，这使得远程过程调用的发生顺序与结果返回顺序无直接关系，而Client端正是通过id识别不同的函数调用。当客户端向服务端发送请求时，只需要填充<strong>id</strong>和<strong>rpcRequest</strong>这两个变量，而剩下的三个变量：<strong>rpcResponse,error,done</strong>,则由服务端根据函数执行情况填充.</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">Call</span> </span>&#123;</span><br><span class="line">  <span class="keyword">final</span> <span class="keyword">int</span> id;               <span class="comment">// call id</span></span><br><span class="line">  <span class="keyword">final</span> <span class="keyword">int</span> retry;           <span class="comment">// retry count</span></span><br><span class="line">  <span class="keyword">final</span> Writable rpcRequest;  <span class="comment">// the serialized rpc request</span></span><br><span class="line">  Writable rpcResponse;       <span class="comment">// null if rpc has error</span></span><br><span class="line">  IOException error;          <span class="comment">// exception, null if success</span></span><br><span class="line">  <span class="keyword">final</span> RPC.RpcKind rpcKind;      <span class="comment">// Rpc EngineKind</span></span><br><span class="line">  <span class="keyword">boolean</span> done;               <span class="comment">// true when call is done	</span></span><br><span class="line">  ...</span><br><span class="line">  <span class="function"><span class="keyword">public</span> <span class="keyword">synchronized</span> <span class="keyword">void</span> <span class="title">setRpcResponse</span><span class="params">(Writable rpcResponse)</span> </span>&#123;</span><br><span class="line">    <span class="keyword">this</span>.rpcResponse = rpcResponse;</span><br><span class="line">    callComplete();</span><br><span class="line">  &#125;</span><br><span class="line">  ...</span><br><span class="line">  <span class="function"><span class="keyword">protected</span> <span class="keyword">synchronized</span> <span class="keyword">void</span> <span class="title">callComplete</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    <span class="keyword">this</span>.done = <span class="keyword">true</span>;</span><br><span class="line">    notify();                                 <span class="comment">// notify caller</span></span><br><span class="line">  &#125;  </span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>通过Call的setRpcResponse来设置RPC请求返回的结果，设置后并调用Call的callComplete方法</p>
</li>
<li><p><strong>private class Connection extends Thread内部类</strong></p>
<p>用Client与每个Server之间维护一个通信连接。该连接相关的基本信息及操作被封装到Connection类中，其中基本信息主要包括：通信连接唯一标识remoteId,与Server端通信的Socket,网络输入流in,网络输出流out,保存RPC请求的哈希表calls等.</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">private</span> <span class="class"><span class="keyword">class</span> <span class="title">Connection</span> <span class="keyword">extends</span> <span class="title">Thread</span> </span>&#123;</span><br><span class="line">    <span class="keyword">private</span> InetSocketAddress server;             <span class="comment">// server ip:port</span></span><br><span class="line">    <span class="keyword">private</span> <span class="keyword">final</span> ConnectionId remoteId;                <span class="comment">// connection id</span></span><br><span class="line">    <span class="keyword">private</span> Socket socket = <span class="keyword">null</span>;                 <span class="comment">// connected socket</span></span><br><span class="line">    <span class="keyword">private</span> DataInputStream in;</span><br><span class="line">    <span class="keyword">private</span> DataOutputStream out;</span><br><span class="line">    ...</span><br><span class="line">    <span class="keyword">private</span> Hashtable&lt;Integer, Call&gt; calls = <span class="keyword">new</span> Hashtable&lt;Integer, Call&gt;();</span><br><span class="line">    ...</span><br><span class="line">    <span class="function"><span class="keyword">private</span> <span class="keyword">synchronized</span> <span class="keyword">void</span> <span class="title">setupIOstreams</span><span class="params">(</span></span></span><br><span class="line"><span class="function"><span class="params">        AtomicBoolean fallbackToSimpleAuth)</span> </span></span><br><span class="line"><span class="function">    </span>&#123;</span><br><span class="line">      <span class="keyword">if</span> (socket != <span class="keyword">null</span> || shouldCloseConnection.get()) &#123;</span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">      &#125; </span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">if</span> (LOG.isDebugEnabled()) &#123;</span><br><span class="line">          LOG.debug(<span class="string">"Connecting to "</span>+server);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">if</span> (Trace.isTracing()) &#123;</span><br><span class="line">          Trace.addTimelineAnnotation(<span class="string">"IPC client connecting to "</span> + server);</span><br><span class="line">        &#125;</span><br><span class="line">        <span class="keyword">short</span> numRetries = <span class="number">0</span>;</span><br><span class="line">        Random rand = <span class="keyword">null</span>;</span><br><span class="line">        <span class="keyword">while</span> (<span class="keyword">true</span>) &#123;</span><br><span class="line">          <span class="comment">// 与远程服务器建立连接, 创建一个Socket对象  </span></span><br><span class="line">          setupConnection();</span><br><span class="line">          InputStream inStream = NetUtils.getInputStream(socket);<span class="comment">// 获取输入流 </span></span><br><span class="line">          OutputStream outStream = NetUtils.getOutputStream(socket); <span class="comment">// 获取输出流</span></span><br><span class="line">          <span class="comment">// 发送RPC Header信息给RPC服务器, 这里RPC服务器正常接收后不会响应, 因为只会验证客户端和服务端RPC程序版本是否匹配, 但是验证没通过后会响应失败状态, 并且服务端会关闭连接 </span></span><br><span class="line">          writeConnectionHeader(outStream);</span><br><span class="line">          ...</span><br><span class="line">          <span class="comment">// 包装输入输出流给in 和 out</span></span><br><span class="line">          <span class="keyword">this</span>.in = <span class="keyword">new</span> DataInputStream(<span class="keyword">new</span> BufferedInputStream(inStream));</span><br><span class="line">          <span class="keyword">if</span> (!(outStream <span class="keyword">instanceof</span> BufferedOutputStream)) &#123;</span><br><span class="line">            outStream = <span class="keyword">new</span> BufferedOutputStream(outStream);</span><br><span class="line">          &#125;</span><br><span class="line">          <span class="keyword">this</span>.out = <span class="keyword">new</span> DataOutputStream(outStream);</span><br><span class="line">          <span class="comment">// 调用start()启动线程</span></span><br><span class="line">          start();</span><br><span class="line">          <span class="keyword">return</span>;</span><br><span class="line">        &#125;</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line">    ...</span><br><span class="line">    </span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>在Connection的setupIOstreams方法中会去建立和服务端的连接，本质会去创建一个Socket对象，建立一个TCP长连接，并且封装相关输入输出流。最后调用start（）启动线程</p>
</li>
</ol>
   <figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">sendRpcRequest</span><span class="params">(<span class="keyword">final</span> Call call)</span></span></span><br><span class="line"><span class="function">        <span class="keyword">throws</span> InterruptedException, IOException </span>&#123;</span><br><span class="line">      <span class="keyword">if</span> (shouldCloseConnection.get()) &#123;</span><br><span class="line">        <span class="keyword">return</span>;</span><br><span class="line">      &#125;</span><br><span class="line">    ...</span><br><span class="line">     <span class="keyword">synchronized</span> (sendRpcRequestLock) &#123;</span><br><span class="line">        Future&lt;?&gt; senderFuture = sendParamsExecutor.submit(<span class="keyword">new</span> Runnable() &#123;</span><br><span class="line">          <span class="meta">@Override</span></span><br><span class="line">          <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">              <span class="keyword">synchronized</span> (Connection.<span class="keyword">this</span>.out) &#123;</span><br><span class="line">                  <span class="comment">//// 对于同一个OutputStream必须同步发送RPC调用, 因为在同一个连接上的多个调用Call必须在同步下进行RPC请求  </span></span><br><span class="line">                <span class="keyword">if</span> (shouldCloseConnection.get()) &#123;</span><br><span class="line">                  <span class="keyword">return</span>;</span><br><span class="line">                &#125;</span><br><span class="line">                </span><br><span class="line">                <span class="keyword">if</span> (LOG.isDebugEnabled())</span><br><span class="line">                  LOG.debug(getName() + <span class="string">" sending #"</span> + call.id);</span><br><span class="line">         </span><br><span class="line">                <span class="keyword">byte</span>[] data = d.getData();</span><br><span class="line">                <span class="keyword">int</span> totalLength = d.getLength();</span><br><span class="line">                out.writeInt(totalLength); <span class="comment">// Total Length 1.写入CallId和调用参数（方法名、方法参数类型、方法参数值）的长度, 4个字节  </span></span><br><span class="line">                out.write(data, <span class="number">0</span>, totalLength);<span class="comment">// RpcRequestHeader + RpcRequest2.写入CallId和序列化后的调用参数（方法名、方法参数类型、方法参数值）  </span></span><br><span class="line">                out.flush();</span><br><span class="line">              &#125;</span><br><span class="line">            &#125; <span class="keyword">catch</span> (IOException e) &#123;...&#125;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">              senderFuture.get();</span><br><span class="line">            &#125; <span class="keyword">catch</span> (ExecutionException e) &#123;</span><br><span class="line">              Throwable cause = e.getCause();&#125;</span><br><span class="line">           ...</span><br><span class="line">          &#125;</span><br><span class="line">        &#125; ...</span><br></pre></td></tr></table></figure>
<p>   客户端发起RPC请求时，会先去把请求相关的调用方法参数等序列化成字节流发送给服务端，核心代码如上</p>
   <figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">      <span class="keyword">if</span> (LOG.isDebugEnabled())</span><br><span class="line">        LOG.debug(getName() + <span class="string">": starting, having connections "</span> </span><br><span class="line">            + connections.size());</span><br><span class="line"></span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">while</span> (waitForWork()) &#123;<span class="comment">//wait here for work - read or close connection</span></span><br><span class="line">          receiveRpcResponse();</span><br><span class="line">        &#125;</span><br><span class="line">      &#125; <span class="keyword">catch</span> (Throwable t) &#123;</span><br><span class="line">        <span class="comment">// This truly is unexpected, since we catch IOException in receiveResponse</span></span><br><span class="line">        <span class="comment">// -- this is only to be really sure that we don't leave a client hanging</span></span><br><span class="line">        <span class="comment">// forever.</span></span><br><span class="line">        LOG.warn(<span class="string">"Unexpected error reading responses on connection "</span> + <span class="keyword">this</span>, t);</span><br><span class="line">        markClosed(<span class="keyword">new</span> IOException(<span class="string">"Error reading responses"</span>, t));</span><br><span class="line">      &#125;</span><br><span class="line">      </span><br><span class="line">      close();</span><br><span class="line">      </span><br><span class="line">      <span class="keyword">if</span> (LOG.isDebugEnabled())</span><br><span class="line">        LOG.debug(getName() + <span class="string">": stopped, remaining connections "</span></span><br><span class="line">            + connections.size());</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>
<p>   connection类的run函数不停地调用receiveRpcResponse（）方法来获取服务端结果</p>
<p>   receiveResponse 函数的关键代码如下，在receiveResponse中主要获取应答头部，根据服务端返回的头部信息判断Rpc请求应答的<strong>status</strong>,并读取<strong>callId</strong>通过callId映射到<strong>Call</strong>对象，并从该Connection持有的所有的calls映射中删除该call，读取输入流，调用Call对象的<strong>setRpcResponse()</strong>为该call设置<strong>RpcResponse</strong></p>
   <figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">receiveRpcResponse</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">        <span class="keyword">int</span> totalLen = in.readInt();</span><br><span class="line">           RpcResponseHeaderProto header = </span><br><span class="line">               RpcResponseHeaderProto.parseDelimitedFrom(in);</span><br><span class="line">           checkResponse(header);</span><br><span class="line">        <span class="keyword">int</span> callId = header.getCallId();</span><br><span class="line">        Call call = calls.get(callId);</span><br><span class="line">        RpcStatusProto status = header.getStatus();</span><br><span class="line">        <span class="keyword">int</span> callId = header.getCallId();</span><br><span class="line">        Call call = calls.get(callId);</span><br><span class="line">        <span class="keyword">if</span> (status == RpcStatusProto.SUCCESS) &#123;</span><br><span class="line">             Writable value = ReflectionUtils.newInstance(valueClass, conf);</span><br><span class="line">             value.readFields(in);                 <span class="comment">// read value</span></span><br><span class="line">             calls.remove(callId);</span><br><span class="line">             call.setRpcResponse(value);</span><br><span class="line">            ...</span><br><span class="line">        &#125;</span><br><span class="line">    &#125;<span class="keyword">catch</span>()&#123;&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<ol start="3">
<li><p>Client 主类</p>
<p><strong>call()方法：</strong>通过ConnectionId获取/建立连接，并封装rpc请求call，通过connection发送rpc请求，发送后同步call代码段中不停地检测call是否done，如果非done则wait()阻塞直到相应的connection调用receiveRpcResponse（）方法触发call.setRpcResponse(value)进而触发callComplete（）方法。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> Writable <span class="title">call</span><span class="params">(RPC.RpcKind rpcKind, Writable rpcRequest,</span></span></span><br><span class="line"><span class="function"><span class="params">     ConnectionId remoteId, <span class="keyword">int</span> serviceClass,</span></span></span><br><span class="line"><span class="function"><span class="params">     AtomicBoolean fallbackToSimpleAuth)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">   <span class="keyword">final</span> Call call = createCall(rpcKind, rpcRequest);</span><br><span class="line">   Connection connection = getConnection(remoteId, call, serviceClass,</span><br><span class="line">     fallbackToSimpleAuth);</span><br><span class="line">   <span class="keyword">try</span> &#123;</span><br><span class="line">     connection.sendRpcRequest(call);                 <span class="comment">// send the rpc request</span></span><br><span class="line">   &#125; <span class="keyword">catch</span> (RejectedExecutionException e) &#123;</span><br><span class="line">     <span class="keyword">throw</span> <span class="keyword">new</span> IOException(<span class="string">"connection has been closed"</span>, e);</span><br><span class="line">   &#125; <span class="keyword">catch</span> (InterruptedException e) &#123;</span><br><span class="line">     Thread.currentThread().interrupt();</span><br><span class="line">     LOG.warn(<span class="string">"interrupted waiting to send rpc request to server"</span>, e);</span><br><span class="line">     <span class="keyword">throw</span> <span class="keyword">new</span> IOException(e);</span><br><span class="line">   &#125;</span><br><span class="line">   </span><br><span class="line">   <span class="keyword">synchronized</span> (call) &#123;</span><br><span class="line">     <span class="keyword">while</span> (!call.done) &#123;</span><br><span class="line">       <span class="keyword">try</span> &#123;</span><br><span class="line">         call.wait();                           <span class="comment">// wait for the result</span></span><br><span class="line">       &#125; <span class="keyword">catch</span> (InterruptedException ie) &#123;</span><br><span class="line">         Thread.currentThread().interrupt();</span><br><span class="line">         <span class="keyword">throw</span> <span class="keyword">new</span> InterruptedIOException(<span class="string">"Call interrupted"</span>);</span><br><span class="line">       &#125;</span><br><span class="line">     &#125;</span><br><span class="line">   </span><br><span class="line">     <span class="keyword">if</span> (call.error != <span class="keyword">null</span>) &#123;</span><br><span class="line">       <span class="keyword">if</span> (call.error <span class="keyword">instanceof</span> RemoteException) &#123;</span><br><span class="line">         call.error.fillInStackTrace();</span><br><span class="line">         <span class="keyword">throw</span> call.error;</span><br><span class="line">       &#125; <span class="keyword">else</span> &#123; <span class="comment">// local exception</span></span><br><span class="line">         InetSocketAddress address = connection.getRemoteAddress();</span><br><span class="line">         <span class="keyword">throw</span> NetUtils.wrapException(address.getHostName(),</span><br><span class="line">                 address.getPort(),</span><br><span class="line">                 NetUtils.getHostname(),</span><br><span class="line">                 <span class="number">0</span>,</span><br><span class="line">                 call.error);</span><br><span class="line">       &#125;</span><br><span class="line">     &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">       <span class="keyword">return</span> call.getRpcResponse();</span><br><span class="line">     &#125;</span><br><span class="line">   &#125;</span><br><span class="line"> &#125;</span><br></pre></td></tr></table></figure>
<p><strong>getConnection()方法：</strong> </p>
<p>首先通过ConnectionID查找client的connections中是否包含改connection, 不包含则创建新的并加入到connections中。调用Connection的setupIOstreams方法包装输入、输出流并调用start().s</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> Connection <span class="title">getConnection</span><span class="params">(ConnectionId remoteId,</span></span></span><br><span class="line"><span class="function"><span class="params">     Call call, <span class="keyword">int</span> serviceClass, AtomicBoolean fallbackToSimpleAuth)</span></span></span><br><span class="line"><span class="function">     <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">   <span class="keyword">if</span> (!running.get()) &#123;</span><br><span class="line">     <span class="comment">// the client is stopped</span></span><br><span class="line">     <span class="keyword">throw</span> <span class="keyword">new</span> IOException(<span class="string">"The client is stopped"</span>);</span><br><span class="line">   &#125;</span><br><span class="line">   Connection connection;</span><br><span class="line">   <span class="comment">/* we could avoid this allocation for each RPC by having a  </span></span><br><span class="line"><span class="comment">    * connectionsId object and with set() method. We need to manage the</span></span><br><span class="line"><span class="comment">    * refs for keys in HashMap properly. For now its ok.</span></span><br><span class="line"><span class="comment">    */</span></span><br><span class="line">   <span class="keyword">do</span> &#123;</span><br><span class="line">     <span class="keyword">synchronized</span> (connections) &#123;</span><br><span class="line">       connection = connections.get(remoteId);</span><br><span class="line">       <span class="keyword">if</span> (connection == <span class="keyword">null</span>) &#123;</span><br><span class="line">         connection = <span class="keyword">new</span> Connection(remoteId, serviceClass);</span><br><span class="line">         connections.put(remoteId, connection);</span><br><span class="line">       &#125;</span><br><span class="line">     &#125;</span><br><span class="line">   &#125; <span class="keyword">while</span> (!connection.addCall(call));</span><br><span class="line">   </span><br><span class="line">   <span class="comment">//we don't invoke the method below inside "synchronized (connections)"</span></span><br><span class="line">   <span class="comment">//block above. The reason for that is if the server happens to be slow,</span></span><br><span class="line">   <span class="comment">//it will take longer to establish a connection and that will slow the</span></span><br><span class="line">   <span class="comment">//entire system down.</span></span><br><span class="line">   connection.setupIOstreams(fallbackToSimpleAuth);</span><br><span class="line">   <span class="keyword">return</span> connection;</span><br><span class="line"> &#125;</span><br></pre></td></tr></table></figure>
</li>
</ol>
<p>综上所述，Client端处理流程具体序列如下图所示：</p>
<p><a href="https://imgchr.com/i/EGAcse" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/04/30/EGAcse.png" alt="EGAcse.png"></a></p>
<h3 id="Server-实现"><a href="#Server-实现" class="headerlink" title="Server 实现"></a>Server 实现</h3><p>Server的结构如下图所示，从图中可以看出，server端包含的内部类比较多，其中一些是和Client端重复的<strong>Call</strong>，还有一些是Server独有的如Reader（Listener的内部类）、Handler、Listener、Responder 他们的作用如下：</p>
<ul>
<li>Listener ： 请求监听类，用于监听客户端发来的请求. </li>
<li>Connection ：连接类，真正的客户端请求读取逻辑在这个类中. </li>
<li>Reader : （Listener的内部类）当监听器监听到用户请求，便让Reader读取用户请求数据. </li>
<li>Call ：用于封装客户端发来的请求. </li>
<li>Handler ：请求处理类，会循环阻塞读取callQueue中的call对象，并对其进行操作.</li>
<li>Responder ：响应RPC请求类，请求处理完毕，由Responder发送给请求客户端. </li>
</ul>
<p><a href="https://imgchr.com/i/EGEVF1" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/04/30/EGEVF1.png" alt="EGEVF1.png"></a></p>
<ol>
<li><p><strong>请求处理阶段</strong></p>
<p>​    该阶段的主要任务是接收来自各个客户端的RPC请求，并将它们封装成固定的格式（Call对象）放到一个共享阻塞队列callQueue中，以便进行后续处理。该阶段内部又分为两个子阶段：请求接收和请求读取，分别有两种线程完成：Listener和Reader请求接收线程Listener初始化源码如下，整个Server只有一个Listener线程，统一负责监听来自客户端的连接请求，一旦有新的请求到达，它会采用轮训的方式从线程池中选择一个Reader线程进行处理。Listener的run() 方法中会阻塞等待客户端请求建立连接，Listener的run()方法的核心代码. </p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">private</span> <span class="class"><span class="keyword">class</span> <span class="title">Listener</span> <span class="keyword">extends</span> <span class="title">Thread</span> </span>&#123;</span><br><span class="line">   </span><br><span class="line">   <span class="keyword">private</span> ServerSocketChannel acceptChannel = <span class="keyword">null</span>; <span class="comment">//the accept channel</span></span><br><span class="line">   <span class="keyword">private</span> Selector selector = <span class="keyword">null</span>; <span class="comment">//the selector that we use for the server</span></span><br><span class="line">   <span class="keyword">private</span> Reader[] readers = <span class="keyword">null</span>;</span><br><span class="line">   <span class="keyword">private</span> <span class="keyword">int</span> currentReader = <span class="number">0</span>;</span><br><span class="line">   <span class="keyword">private</span> InetSocketAddress address; <span class="comment">//the address we bind at</span></span><br><span class="line">   ...</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Listerner 的run方法: 在Selector中</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">     LOG.info(Thread.currentThread().getName() + <span class="string">": starting"</span>);</span><br><span class="line">     SERVER.set(Server.<span class="keyword">this</span>);</span><br><span class="line">     connectionManager.startIdleScan();</span><br><span class="line">     <span class="keyword">while</span> (running) &#123;</span><br><span class="line">       SelectionKey key = <span class="keyword">null</span>;</span><br><span class="line">       <span class="keyword">try</span> &#123;</span><br><span class="line">         getSelector().select();<span class="comment">// 如果Selector中注册的ServerSocketChannel没有新的Socket请求的话, 就阻塞在这里  </span></span><br><span class="line">         Iterator&lt;SelectionKey&gt; iter = getSelector().selectedKeys().iterator();</span><br><span class="line">         <span class="keyword">while</span> (iter.hasNext()) &#123;</span><br><span class="line">           key = iter.next();</span><br><span class="line">           iter.remove();</span><br><span class="line">           <span class="keyword">try</span> &#123;</span><br><span class="line">             <span class="keyword">if</span> (key.isValid()) &#123;</span><br><span class="line">               <span class="keyword">if</span> (key.isAcceptable())</span><br><span class="line">                 doAccept(key);<span class="comment">// 处理连接事件  </span></span><br><span class="line">             &#125;</span><br><span class="line">           &#125; <span class="keyword">catch</span> (IOException e) &#123;</span><br><span class="line">           &#125;</span><br><span class="line">           key = <span class="keyword">null</span>;</span><br><span class="line">         &#125;</span><br><span class="line">       &#125; <span class="keyword">catch</span> (OutOfMemoryError e) &#123;</span><br><span class="line">         <span class="comment">// we can run out of memory if we have too many threads</span></span><br><span class="line">         <span class="comment">// log the event and sleep for a minute and give </span></span><br><span class="line">         <span class="comment">// some thread(s) a chance to finish</span></span><br><span class="line">         LOG.warn(<span class="string">"Out of Memory in server select"</span>, e);</span><br><span class="line">         closeCurrentConnection(key, e);</span><br><span class="line">         connectionManager.closeIdle(<span class="keyword">true</span>);</span><br><span class="line">         <span class="keyword">try</span> &#123; Thread.sleep(<span class="number">60000</span>); &#125; <span class="keyword">catch</span> (Exception ie) &#123;&#125;</span><br><span class="line">       &#125; <span class="keyword">catch</span> (Exception e) &#123;</span><br><span class="line">         closeCurrentConnection(key, e);</span><br><span class="line">       &#125;</span><br><span class="line">     &#125;</span><br><span class="line">    ...</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>紧接着具体的请求接收处理是在Listener的doAccept()方法中处理的，获取连接后会往Reader线程中的多路复用器Selector注册连接，Listener的doAccept方法的核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">void</span> <span class="title">doAccept</span><span class="params">(SelectionKey key)</span> <span class="keyword">throws</span> InterruptedException, IOException,  OutOfMemoryError </span>&#123;</span><br><span class="line">      ServerSocketChannel server = (ServerSocketChannel) key.channel();<span class="comment">// 拿到ServerSocketchannel  </span></span><br><span class="line">      SocketChannel channel;<span class="comment">// 拿到Socketchannel  </span></span><br><span class="line">      <span class="keyword">while</span> ((channel = server.accept()) != <span class="keyword">null</span>) &#123; <span class="comment">// 非阻塞的拿到SocketChannel  </span></span><br><span class="line"></span><br><span class="line">        channel.configureBlocking(<span class="keyword">false</span>);<span class="comment">// 把SocketChannel设置为非阻塞模式  </span></span><br><span class="line">        channel.socket().setTcpNoDelay(tcpNoDelay);</span><br><span class="line">        channel.socket().setKeepAlive(<span class="keyword">true</span>);</span><br><span class="line">        </span><br><span class="line">        Reader reader = getReader();<span class="comment">// 随机轮询获取一个Rearder线程  </span></span><br><span class="line">        Connection c = connectionManager.register(channel);</span><br><span class="line">        <span class="comment">// If the connectionManager can't take it, close the connection.</span></span><br><span class="line">        <span class="keyword">if</span> (c == <span class="keyword">null</span>) &#123;</span><br><span class="line">          <span class="keyword">if</span> (channel.isOpen()) &#123;</span><br><span class="line">            IOUtils.cleanup(<span class="keyword">null</span>, channel);</span><br><span class="line">          &#125;</span><br><span class="line">          connectionManager.droppedConnections.getAndIncrement();</span><br><span class="line">          <span class="keyword">continue</span>;</span><br><span class="line">        &#125;</span><br><span class="line">        key.attach(c);  <span class="comment">// so closeCurrentConnection can get the object</span></span><br><span class="line">        reader.addConnection(c);</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>
<p>​    客户端和服务端连接建立成功之后，服务端的Reader线程中维护了连接，有了连接就可以传输数据，Reader线程的run方法中就是阻塞去等待客户端的请求数据，一旦该连接上有可读数据，该Reader线程就会被唤醒，紧接着会去解析字节流序列化请求数据，封装成Call对象，塞到callQueue阻塞队列，Reader的run()方法的核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">       LOG.info(<span class="string">"Starting "</span> + Thread.currentThread().getName());</span><br><span class="line">       <span class="keyword">try</span> &#123;</span><br><span class="line">         doRunLoop();</span><br><span class="line">       &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">         <span class="keyword">try</span> &#123;</span><br><span class="line">           readSelector.close();</span><br><span class="line">         &#125; <span class="keyword">catch</span> (IOException ioe) &#123;</span><br><span class="line">           LOG.error(<span class="string">"Error closing read selector in "</span> + Thread.currentThread().getName(), ioe);</span><br><span class="line">         &#125;</span><br><span class="line">       &#125;</span><br><span class="line">     &#125;</span><br><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">synchronized</span> <span class="keyword">void</span> <span class="title">doRunLoop</span><span class="params">()</span> </span>&#123;</span><br><span class="line">       <span class="keyword">while</span> (running) &#123;</span><br><span class="line">         SelectionKey key = <span class="keyword">null</span>;</span><br><span class="line">         <span class="keyword">try</span> &#123;</span><br><span class="line">           <span class="comment">// consume as many connections as currently queued to avoid</span></span><br><span class="line">           <span class="comment">// unbridled acceptance of connections that starves the select</span></span><br><span class="line">           <span class="keyword">int</span> size = pendingConnections.size();</span><br><span class="line">           <span class="keyword">for</span> (<span class="keyword">int</span> i=size; i&gt;<span class="number">0</span>; i--) &#123;</span><br><span class="line">             Connection conn = pendingConnections.take();</span><br><span class="line">             conn.channel.register(readSelector, SelectionKey.OP_READ, conn);</span><br><span class="line">           &#125;</span><br><span class="line">           readSelector.select();<span class="comment">// 如果Selector中注册的SocketChannel中都没有可读数据的话, 就阻塞在这里  </span></span><br><span class="line">   </span><br><span class="line">           Iterator&lt;SelectionKey&gt; iter = readSelector.selectedKeys().iterator();</span><br><span class="line">           <span class="keyword">while</span> (iter.hasNext()) &#123;</span><br><span class="line">             key = iter.next();</span><br><span class="line">             iter.remove();</span><br><span class="line">             <span class="keyword">try</span> &#123;</span><br><span class="line">               <span class="keyword">if</span> (key.isReadable()) &#123; <span class="comment">// SocketChannel有可读数据  </span></span><br><span class="line">                 doRead(key);</span><br><span class="line">               &#125;</span><br><span class="line">                 ...</span><br><span class="line">             &#125;</span><br><span class="line">           &#125;</span><br><span class="line">         &#125;</span><br><span class="line">       &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>​    在Reader的run 中调用了doRunLoop()方法，该方法将connections注册到readSelector，并调用doRead()读取SockletChannel中的数据（如果有）。doRead（）中具体的读取及解析请求数据交给Connection来处理，核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br></pre></td><td class="code"><pre><span class="line"> <span class="function"><span class="keyword">void</span> <span class="title">doRead</span><span class="params">(SelectionKey key)</span> <span class="keyword">throws</span> InterruptedException </span>&#123;</span><br><span class="line">      <span class="keyword">int</span> count = <span class="number">0</span>;</span><br><span class="line">      Connection c = (Connection)key.attachment();</span><br><span class="line">      <span class="keyword">if</span> (c == <span class="keyword">null</span>) &#123;</span><br><span class="line">        <span class="keyword">return</span>;  </span><br><span class="line">      &#125;</span><br><span class="line">      c.setLastContact(Time.now());</span><br><span class="line">      </span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        count = c.readAndProcess();</span><br><span class="line">      &#125; <span class="keyword">catch</span> (InterruptedException ieo) &#123;</span><br><span class="line">        LOG.info(Thread.currentThread().getName() + <span class="string">": readAndProcess caught InterruptedException"</span>, ieo);</span><br><span class="line">        <span class="keyword">throw</span> ieo;</span><br><span class="line">      &#125; <span class="keyword">catch</span> (Exception e) &#123;...</span><br><span class="line">                            &#125;</span><br><span class="line"> &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>在doRead中调用了Connection的readAndProcess（）方法，接着来看Connection类的readAndProcess()方法，主要从连接中读取请求数据，核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">int</span> <span class="title">readAndProcess</span><span class="params">()</span></span></span><br><span class="line"><span class="function">       <span class="keyword">throws</span> WrappedRpcServerException, IOException, InterruptedException </span>&#123;</span><br><span class="line">     <span class="keyword">while</span> (<span class="keyword">true</span>) &#123;</span><br><span class="line">         ...;</span><br><span class="line">        <span class="keyword">if</span> (data == <span class="keyword">null</span>) &#123;</span><br><span class="line">         dataLengthBuffer.flip();</span><br><span class="line">         dataLength = dataLengthBuffer.getInt();</span><br><span class="line">         checkDataLength(dataLength);</span><br><span class="line">         data = ByteBuffer.allocate(dataLength);<span class="comment">// 根据dataLength创建一个dataLength大小的缓冲区, 用来读数据  </span></span><br><span class="line">       &#125;</span><br><span class="line">       </span><br><span class="line">       count = channelRead(channel, data);<span class="comment">// 读取第一次请求Header信息或请求数据  </span></span><br><span class="line">       </span><br><span class="line">       <span class="keyword">if</span> (data.remaining() == <span class="number">0</span>) &#123;</span><br><span class="line">         dataLengthBuffer.clear();<span class="comment">// 清空dataLengthBuffer  </span></span><br><span class="line">         data.flip();</span><br><span class="line">         <span class="keyword">boolean</span> isHeaderRead = connectionContextRead;</span><br><span class="line">         processOneRpc(data.array());<span class="comment">// 处理rpc请求,把封装好的请求信息Call塞到callQueue阻塞队列  </span></span><br><span class="line">         data = <span class="keyword">null</span>;</span><br><span class="line">         <span class="keyword">if</span> (!isHeaderRead) &#123; <span class="comment">// 读取第一次RPC请求Header之后会再continue, 继续读取请求数据</span></span><br><span class="line">           <span class="keyword">continue</span>;</span><br><span class="line">         &#125;</span><br><span class="line">       &#125; </span><br><span class="line">       <span class="keyword">return</span> count;</span><br><span class="line">     &#125;</span><br></pre></td></tr></table></figure>
<p>在readAndProcess中调用processOneRpc()方法处理rpc请求，在processOneRpc（）中调用processRpcRequest（）方法来将请求解析封装成server端的Call对象并加入callQueue中。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">processOneRpc</span><span class="params">(<span class="keyword">byte</span>[] buf)</span></span></span><br><span class="line"><span class="function">    <span class="keyword">throws</span> IOException, WrappedRpcServerException, InterruptedException </span>&#123;</span><br><span class="line">    <span class="keyword">int</span> callId = -<span class="number">1</span>;</span><br><span class="line">    <span class="comment">// 通过输入流读取buf</span></span><br><span class="line">    <span class="keyword">final</span> DataInputStream dis =</span><br><span class="line">            <span class="keyword">new</span> DataInputStream(<span class="keyword">new</span> ByteArrayInputStream(buf));</span><br><span class="line">    <span class="comment">// 通过流操作，获取header</span></span><br><span class="line">    <span class="keyword">final</span> RpcRequestHeaderProto header =</span><br><span class="line">            decodeProtobufFromStream(RpcRequestHeaderProto.newBuilder(), dis);</span><br><span class="line">    callId = header.getCallId();</span><br><span class="line">    callId = header.getCallId();</span><br><span class="line">    retry = header.getRetryCount();</span><br><span class="line">    .......;</span><br><span class="line">    <span class="keyword">if</span> (callId &lt; <span class="number">0</span>) &#123; <span class="comment">// callIds typically used during connection setup</span></span><br><span class="line">          processRpcOutOfBandRequest(header, dis);</span><br><span class="line">        &#125; <span class="keyword">else</span> <span class="keyword">if</span> (!connectionContextRead) &#123;</span><br><span class="line">          <span class="keyword">throw</span> <span class="keyword">new</span> WrappedRpcServerException(</span><br><span class="line">              RpcErrorCodeProto.FATAL_INVALID_RPC_HEADER,</span><br><span class="line">              <span class="string">"Connection context not established"</span>);</span><br><span class="line">        &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">        <span class="comment">// callId正常 调用processRpcRequest</span></span><br><span class="line">          processRpcRequest(header, dis);</span><br><span class="line">        &#125;</span><br><span class="line">      &#125; <span class="keyword">catch</span> (WrappedRpcServerException wrse) &#123; <span class="comment">// inform client of error</span></span><br><span class="line">        Throwable ioe = wrse.getCause();</span><br><span class="line">    	<span class="comment">//构造error call，并调用setupResponse函数通知给客户端错误。</span></span><br><span class="line">        <span class="keyword">final</span> Call call = <span class="keyword">new</span> Call(callId, retry, <span class="keyword">null</span>, <span class="keyword">this</span>);</span><br><span class="line">        setupResponse(authFailedResponse, call,</span><br><span class="line">            RpcStatusProto.FATAL, wrse.getRpcErrorCodeProto(), <span class="keyword">null</span>,</span><br><span class="line">            ioe.getClass().getName(), ioe.getMessage());</span><br><span class="line">        call.sendResponse();</span><br><span class="line">        <span class="keyword">throw</span> wrse;</span><br><span class="line">      &#125;</span><br></pre></td></tr></table></figure>
</li>
</ol>
   <figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">processRpcRequest</span><span class="params">(RpcRequestHeaderProto header,</span></span></span><br><span class="line"><span class="function"><span class="params">       DataInputStream dis)</span> <span class="keyword">throws</span> WrappedRpcServerException,</span></span><br><span class="line"><span class="function">       InterruptedException </span>&#123;</span><br><span class="line">     Writable rpcRequest;</span><br><span class="line">     <span class="keyword">try</span> &#123; <span class="comment">//Read the rpc request</span></span><br><span class="line">       rpcRequest = ReflectionUtils.newInstance(rpcRequestClass, conf);</span><br><span class="line">       rpcRequest.readFields(dis);</span><br><span class="line">     &#125;<span class="keyword">catch</span>()&#123;&#125;</span><br><span class="line">          ......;</span><br><span class="line">  <span class="comment">// 构造新call</span></span><br><span class="line">     Call call = <span class="keyword">new</span> Call(header.getCallId(), header.getRetryCount(),</span><br><span class="line">         rpcRequest, <span class="keyword">this</span>, ProtoUtil.convert(header.getRpcKind()),</span><br><span class="line">         header.getClientId().toByteArray(), traceSpan);</span><br><span class="line">     <span class="comment">//将call 加入到队列中</span></span><br><span class="line">      <span class="keyword">if</span> (callQueue.isClientBackoffEnabled()) &#123;</span><br><span class="line">       <span class="comment">// if RPC queue is full, we will ask the RPC client to back off by</span></span><br><span class="line">       <span class="comment">// throwing RetriableException. Whether RPC client will honor</span></span><br><span class="line">       <span class="comment">// RetriableException and retry depends on client ipc retry policy.</span></span><br><span class="line">       <span class="comment">// For example, FailoverOnNetworkExceptionRetry handles</span></span><br><span class="line">       <span class="comment">// RetriableException.</span></span><br><span class="line">       queueRequestOrAskClientToBackOff(call);</span><br><span class="line">     &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">       callQueue.put(call);              <span class="comment">// queue the call; maybe blocked here</span></span><br><span class="line">     &#125;</span><br><span class="line">     incRpcCount();  <span class="comment">// Increment the rpc count</span></span><br><span class="line">     </span><br><span class="line">           </span><br><span class="line">       &#125;</span><br></pre></td></tr></table></figure>
<p>   <strong>至此请求接收结束。</strong></p>
<hr>
<ol start="2">
<li><p><strong>请求处理</strong></p>
<p>该阶段的主要任务是从共享队列callQueue中获取Call对象，执行相应的函数调用，并将结果返回给客户端，这全部由Handler线程完成的。Server端可同时存在多个Handler线程。它们并行从共享队列中读取Call对象,经执行对应的韩式调用后，将尝试着直接将结果返回给对应的客户端。但考虑到某些函数调用返回的结果很大或者网络速度过慢，可能难以将结果一次性发送到客户端，此时Handler将尝试着将后续发送任务交给Responder线程。Handler的run方法中会阻塞等待callQueue队列中有请求数据，Handler的run()核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123; </span><br><span class="line">    LOG.debug(Thread.currentThread().getName() + <span class="string">": starting"</span>);</span><br><span class="line">      SERVER.set(Server.<span class="keyword">this</span>);</span><br><span class="line">      ByteArrayOutputStream buf = </span><br><span class="line">        <span class="keyword">new</span> ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);</span><br><span class="line">    <span class="keyword">while</span> (running) &#123;</span><br><span class="line">            TraceScope traceScope = <span class="keyword">null</span>;</span><br><span class="line">            <span class="keyword">try</span> &#123;</span><br><span class="line">              <span class="keyword">final</span> Call call = callQueue.take(); <span class="comment">// pop the queue; maybe blocked here</span></span><br><span class="line">              String errorClass = <span class="keyword">null</span>;</span><br><span class="line">              String error = <span class="keyword">null</span>;</span><br><span class="line">              RpcStatusProto returnStatus = RpcStatusProto.SUCCESS;</span><br><span class="line">              RpcErrorCodeProto detailedErr = <span class="keyword">null</span>;</span><br><span class="line">              Writable value = <span class="keyword">null</span>;</span><br><span class="line"></span><br><span class="line">              CurCall.set(call);</span><br><span class="line">              ...;</span><br><span class="line"></span><br><span class="line">              <span class="keyword">try</span> &#123;</span><br><span class="line">                <span class="comment">// Make the call as the user via Subject.doAs, thus associating</span></span><br><span class="line">                <span class="comment">// the call with the Subject</span></span><br><span class="line">                <span class="keyword">if</span> (call.connection.user == <span class="keyword">null</span>) &#123;</span><br><span class="line">                  value = call(call.rpcKind, call.connection.protocolName, call.rpcRequest, call.timestamp);</span><br><span class="line">                &#125; <span class="keyword">else</span> &#123;</span><br><span class="line">                  value = </span><br><span class="line">                    call.connection.user.doAs</span><br><span class="line">                      (<span class="keyword">new</span> PrivilegedExceptionAction&lt;Writable&gt;() &#123;</span><br><span class="line">                         <span class="meta">@Override</span></span><br><span class="line">                         <span class="function"><span class="keyword">public</span> Writable <span class="title">run</span><span class="params">()</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">                           <span class="comment">// make the call</span></span><br><span class="line">                          <span class="comment">// 反射调用对应服务，返回结果ObjectWritable, 传入Connection中接口的Class对象, 是在建立连接之后第一次客户端请求带过来的  </span></span><br><span class="line">                           <span class="keyword">return</span> call(call.rpcKind, call.connection.protocolName, </span><br><span class="line">                                       call.rpcRequest, call.timestamp);</span><br><span class="line"></span><br><span class="line">                         &#125;</span><br><span class="line">                       &#125;</span><br><span class="line">                      );</span><br><span class="line">                &#125;<span class="keyword">catch</span>(Expection e)&#123;...&#125;</span><br><span class="line">                ...;</span><br><span class="line">                 CurCall.set(<span class="keyword">null</span>);</span><br><span class="line">              <span class="keyword">synchronized</span> (call.connection.responseQueue) &#123;</span><br><span class="line">                  <span class="comment">// 同一个连接上的多个响应必须在同步下进行  </span></span><br><span class="line">                setupResponse(buf, call, returnStatus, detailedErr,</span><br><span class="line">                    value, errorClass, error);<span class="comment">// 生成返回给客户端的数据包,包含(客户端调用ID+状态status+RPC方法返回值),设置到Call对象中  </span></span><br><span class="line"></span><br><span class="line">                <span class="comment">// Discard the large buf and reset it back to smaller size</span></span><br><span class="line">                <span class="comment">// to free up heap.</span></span><br><span class="line">                <span class="keyword">if</span> (buf.size() &gt; maxRespSize) &#123;</span><br><span class="line">                  LOG.warn(<span class="string">"Large response size "</span> + buf.size() + <span class="string">" for call "</span></span><br><span class="line">                      + call.toString());</span><br><span class="line">                  buf = <span class="keyword">new</span> ByteArrayOutputStream(INITIAL_RESP_BUF_SIZE);</span><br><span class="line">                &#125;</span><br><span class="line">                call.sendResponse();</span><br><span class="line">              &#125;<span class="keyword">catch</span>(Exception e)&#123;...&#125;</span><br><span class="line"></span><br><span class="line">            &#125;</span><br><span class="line">     &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>
<p>服务端拿到调用参数之后，会反射调用对应服务，返回方法返回值</p>
</li>
</ol>
<ol start="3">
<li><p><strong>请求响应</strong></p>
<p>​    每个Handler线程执行完函数调用后，会尝试着将执行结果返回给客户端，但由于特殊情况，比如函数调用返回的结果过大或者网络异常情况，会将发送任务交给Responder线程，Server端仅存在一个Responder线程，它的内部包含一个多路复用器Selector对象，用于监听SelectionKey.OP_WRITE事件，当Handler没能够将结果一次性发送到客户端时，会向该Selector对象注册SelectorKey.OP_WRITE事件，进而由Responder线程采用异步方式继续发送未发送完成的结果，具体的核心代码如下：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">run</span><span class="params">()</span> </span>&#123;</span><br><span class="line">      LOG.info(Thread.currentThread().getName() + <span class="string">": starting"</span>);</span><br><span class="line">      SERVER.set(Server.<span class="keyword">this</span>);</span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        doRunLoop();</span><br><span class="line">      &#125; <span class="keyword">finally</span> &#123;</span><br><span class="line">        LOG.info(<span class="string">"Stopping "</span> + Thread.currentThread().getName());</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">          writeSelector.close();</span><br><span class="line">        &#125; <span class="keyword">catch</span> (IOException ioe) &#123;</span><br><span class="line">          LOG.error(<span class="string">"Couldn't close write selector in "</span> + Thread.currentThread().getName(), ioe);</span><br><span class="line">        &#125;</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br></pre></td></tr></table></figure>
<p>看看 doRunLoop函数干什么,从多路复用器Selector对象获取Handler 未发送的结果，调用doAsyncWrite异步写发送。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">doRunLoop</span><span class="params">()</span> </span>&#123;</span><br><span class="line">  <span class="keyword">long</span> lastPurgeTime = <span class="number">0</span>;   <span class="comment">// last check for old calls.</span></span><br><span class="line"></span><br><span class="line">  <span class="keyword">while</span> (running) &#123;</span><br><span class="line">    <span class="keyword">try</span> &#123;</span><br><span class="line">      waitPending();     <span class="comment">// If a channel is being registered, wait.</span></span><br><span class="line">      writeSelector.select(PURGE_INTERVAL);</span><br><span class="line">      Iterator&lt;SelectionKey&gt; iter = writeSelector.selectedKeys().iterator();</span><br><span class="line">      <span class="keyword">while</span> (iter.hasNext()) &#123;</span><br><span class="line">        SelectionKey key = iter.next();</span><br><span class="line">        iter.remove();</span><br><span class="line">        <span class="keyword">try</span> &#123;</span><br><span class="line">          <span class="keyword">if</span> (key.isWritable()) &#123;</span><br><span class="line">            doAsyncWrite(key);</span><br><span class="line">          &#125;</span><br><span class="line">        &#125; <span class="keyword">catch</span> (CancelledKeyException cke) &#123;...&#125;</span><br><span class="line">        ...;</span><br><span class="line"></span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>那再看看doAsyncWrite（）内部</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"><span class="function"><span class="keyword">private</span> <span class="keyword">void</span> <span class="title">doAsyncWrite</span><span class="params">(SelectionKey key)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">  Call call = (Call)key.attachment();</span><br><span class="line">  <span class="keyword">if</span> (call == <span class="keyword">null</span>) &#123;</span><br><span class="line">    <span class="keyword">return</span>;</span><br><span class="line">  &#125;</span><br><span class="line">  <span class="keyword">if</span> (key.channel() != call.connection.channel) &#123;</span><br><span class="line">    <span class="keyword">throw</span> <span class="keyword">new</span> IOException(<span class="string">"doAsyncWrite: bad channel"</span>);</span><br><span class="line">  &#125;</span><br><span class="line"></span><br><span class="line">  <span class="keyword">synchronized</span>(call.connection.responseQueue) &#123;<span class="comment">//  同一个连接上的多个响应必须在同步下进行</span></span><br><span class="line">    <span class="keyword">if</span> (processResponse(call.connection.responseQueue, <span class="keyword">false</span>)) &#123;</span><br><span class="line">      <span class="keyword">try</span> &#123;</span><br><span class="line">        key.interestOps(<span class="number">0</span>);</span><br><span class="line">      &#125; <span class="keyword">catch</span> (CancelledKeyException e) &#123;</span><br><span class="line">        <span class="comment">/* The Listener/reader might have closed the socket.</span></span><br><span class="line"><span class="comment">         * We don't explicitly cancel the key, so not sure if this will</span></span><br><span class="line"><span class="comment">         * ever fire.</span></span><br><span class="line"><span class="comment">         * This warning could be removed.</span></span><br><span class="line"><span class="comment">         */</span></span><br><span class="line">        LOG.warn(<span class="string">"Exception while changing ops : "</span> + e);</span><br><span class="line">      &#125;</span><br><span class="line">    &#125;</span><br><span class="line">  &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<p>Server 端的状态转移图如下所示：</p>
<p><a href="https://imgchr.com/i/EWF8gI" target="_blank" rel="noopener"><img src="https://s2.ax1x.com/2019/05/10/EWF8gI.md.png" alt="EWF8gI.md.png"></a></p>
</li>
</ol>


        
        <div class="post-announce">
            感谢您的阅读，本文由
            <a href="https://spaces-x.github.io">Space-X</a>
            版权所有。如若转载，请注明出处：Space-X（<a href="https://spaces-x.github.io/2019/05/10/hadoopRPC/">https://spaces-x.github.io/2019/05/10/hadoopRPC/</a>）
        </div>
        

        <div class="post__prevs">
            <div class="post__prev">
                
                <a href="/2019/04/04/memorypaging/" title="Memory Paging"><i class="iconfont icon-prev"></i>Memory Paging</a>
                
            </div>
            <div class="post__prev post__prev--right">
                
            </div>
        </div>
    </div>
</article>

        
        
    </div>

    <aside class="page__sidebar">
    <!-- 
        <div class="sidebar__img">
            <img src="https://s1.ax1x.com/2018/08/08/PsWmg1.jpg" alt="Space-X" title="Space-X">
        </div>
     -->

    <form id="page-search-from" class="page__search-from" action="/search/">
        <label class="search-form__item">
            <input class="input" type="text" name="search" placeholder="Search...">
            <i class="iconfont icon-search"></i>
        </label>
    </form>

    
        <div class="sidebar__block">
            <h3 class="block__title">简介</h3>
            <p class="block__text">日常学习与兴趣交流的个人博客</p>
        </div>
    

    <div class="sidebar__block">
        <h3 class="block__title">文章分类</h3>
        <ul class="block-list"><li class="block-list-item"><a class="block-list-link" href="/categories/计算机网络/">计算机网络</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/算法/">算法</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/游玩/">游玩</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/welcome/">welcome</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/java设计模式/">java设计模式</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/hadoop/">hadoop</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/RPC/">RPC</a><span class="block-list-count">1</span><ul class="block-list-child"><li class="block-list-item"><a class="block-list-link" href="/categories/RPC/Hadoop/">Hadoop</a><span class="block-list-count">1</span></li></ul></li><li class="block-list-item"><a class="block-list-link" href="/categories/OS/">OS</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/Hbase/">Hbase</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/Hadoop/">Hadoop</a><span class="block-list-count">2</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/Git-版本管理/">Git 版本管理</a><span class="block-list-count">1</span></li><li class="block-list-item"><a class="block-list-link" href="/categories/GFW/">GFW</a><span class="block-list-count">1</span></li></ul>
    </div>
    
    <div class="sidebar__block">
        <h3 class="block__title">最新文章</h3>
        
        <ul class="block-list latest-post-list">
            
                    <li class="latest-post-item">
                        <a href="/2019/05/10/hadoopRPC/" title="hadoopRPC">
                            <div class="item__cover">
                                <img src="https://s2.ax1x.com/2019/05/10/EWAoB6.md.jpg" alt="hadoopRPC" />
                            </div>
                            <div class="item__info">
                                <h3 class="item__title">hadoopRPC</h3>
                                <span class="item__text">2019-05-10</span>
                            </div>
                        </a>
                    </li>
                
                    <li class="latest-post-item">
                        <a href="/2019/04/04/memorypaging/" title="Memory Paging">
                            <div class="item__cover">
                                <img src="https://s2.ax1x.com/2019/04/04/A2ifUO.md.jpg" alt="Memory Paging" />
                            </div>
                            <div class="item__info">
                                <h3 class="item__title">Memory Paging</h3>
                                <span class="item__text">2019-04-04</span>
                            </div>
                        </a>
                    </li>
                
                    <li class="latest-post-item">
                        <a href="/2019/03/31/DecoratorMode/" title="DecoratorMode">
                            <div class="item__cover">
                                <img src="https://s2.ax1x.com/2019/03/31/ArQrQJ.md.jpg" alt="DecoratorMode" />
                            </div>
                            <div class="item__info">
                                <h3 class="item__title">DecoratorMode</h3>
                                <span class="item__text">2019-03-31</span>
                            </div>
                        </a>
                    </li>
                
                    <li class="latest-post-item">
                        <a href="/2019/01/18/alogorithm/" title="Algorithm1">
                            <div class="item__cover">
                                <img src="https://s2.ax1x.com/2019/01/18/k9gwCD.jpg" alt="Algorithm1" />
                            </div>
                            <div class="item__info">
                                <h3 class="item__title">Algorithm1</h3>
                                <span class="item__text">2019-01-18</span>
                            </div>
                        </a>
                    </li>
                
        </ul>
    
    </div>

    <div class="sidebar__block">
        <h3 class="block__title">文章标签</h3>
        
        <ul class="block-list tag-list clearfix">
            
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Eclipse/">Eclipse</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Google-PageRank/">Google,PageRank</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Great-Firewall/">Great Firewall</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Hadoop/">Hadoop</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Hadoop部署/">Hadoop部署</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Hbase/">Hbase</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Learning/">Learning</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Life/">Life</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/Map-Reduce/">Map-Reduce</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/RPC/">RPC</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/TCP-Socket通信/">TCP,Socket通信</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/git/">git</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/hadoop/">hadoop</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/hobbies/">hobbies</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/java/">java</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/linux/">linux</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/love/">love</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/memory/">memory</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/noSQL/">noSQL</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/paging/">paging</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/shadowsocks/">shadowsocks</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/travel/">travel</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/分支定界/">分支定界</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/列式数据库/">列式数据库</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/动态规划/">动态规划</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/版本管理/">版本管理</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/装饰器模式/">装饰器模式</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/计算机网络/">计算机网络</a>
                    </li>  
                
                    <li class="tag-item">
                        <a class="tag-link" href="/tags/读取数据/">读取数据</a>
                    </li>  
                
        </ul>
    
    </div>

    <!-- <div class="sidebar__block">
        <h3 class="block__title">友情链接</h3>
        <ul class="block-list">
            
        </ul>
    </div> -->
</aside>
</main>


        
            <!-- 页脚 -->
<footer class="page__footer">
    <section class="footer__top">
        <div class="page__container footer__container">
            
            <div class="footer-top__item footer-top__item--2">
                <h3 class="item__title">关于</h3>
                <div class="item__content">
                    <p class="item__text">本站是基于 Hexo 搭建的静态资源博客，主要用于分享日常学习、生活及工作的一些心得总结，欢迎点击右下角订阅 rss。</p>
                    <ul class="footer__contact-info">
                        <li class="contact-info__item">
                            <i class="iconfont icon-address"></i>
                            <span>Dalian, Lianning Province, China</span>
                        </li>
                        <li class="contact-info__item">
                            <i class="iconfont icon-email2"></i>
                            <span>dlut.weixiang@gmail.com</span>
                        </li>
                    </ul>
                </div>
            </div>

            
            
                <div class="footer-top__item footer__image">
                    <img src="https://s1.ax1x.com/2018/08/08/PsWmg1.jpg" alt="logo" title="Space-X">
                </div>
            
            
            
            
                
                    <div class="footer-top__item">
                        <h3 class="item__title">友情链接</h3>
                        <div class="item__content">
                            <ul class="footer-top__list">
                                
                                    <li class="list-item">
                                        <a href="https://github.com/Mrminfive/hexo-theme-skapp" title="hexo-theme-skapp" target="_blank">hexo-theme-skapp</a>
                                    </li>
                                
                            </ul>
                        </div>
                    </div>
                
                    <div class="footer-top__item">
                        <h3 class="item__title">构建工具</h3>
                        <div class="item__content">
                            <ul class="footer-top__list">
                                
                                    <li class="list-item">
                                        <a href="https://hexo.io/" title="Blog Framework" target="_blank">Hexo</a>
                                    </li>
                                
                            </ul>
                        </div>
                    </div>
                
            
        </div>
    </section>
    <section class="footer__bottom">
        <div class="page__container footer__container">
            <p class="footer__copyright">©
                <a href="https://github.com/Mrminfive/hexo-theme-skapp" target="_blank">Skapp</a> 2017 powered by
                <a href="http://hexo.io/" target="_blank">Hexo</a>, made by 
                <a href="https://github.com/Mrminfive" target="_blank">minfive</a>.
            </p>
            <ul class="footer__social-network clearfix">
                
                
                    <li class="social-network__item">
                        <a href="https://github.com/spaces-x" target="_blank" title="github">
                            <i class="iconfont icon-github"></i>
                        </a>
                    </li>
                
                    <li class="social-network__item">
                        <a href="mailto:dlut.weixiang@gmail.com" target="_blank" title="email">
                            <i class="iconfont icon-email"></i>
                        </a>
                    </li>
                
                    <li class="social-network__item">
                        <a href="/atom.xml" target="_blank" title="rss">
                            <i class="iconfont icon-rss"></i>
                        </a>
                    </li>
                
                
            </ul>
        </div>
    </section>
</footer>
        

        
            <!-- 返回顶部 -->
<div id="back-top" class="back-top back-top--hidden js-hidden">
    <i class="iconfont icon-top"></i>
</div>
        
    </div>

    <!-- build:js /js/common.js -->
        <script type="text/javascript" src="js/common/utils.js"></script>
        <script type="text/javascript" src="js/common/pack.js"></script>
        <script type="text/javascript" src="js/common/animation.js"></script>
        <script type="text/javascript" src="js/layout/loading.js"></script>
        <script type="text/javascript" src="js/layout/header.js"></script>
        <script type="text/javascript" src="js/layout/back-top.js"></script>
        <script type="text/javascript" src="js/layout/post.js"></script>
    <!-- endbuild -->

    
    <script src="/js/page/post.js"></script>


    
    



    <!-- 不蒜子统计 -->

<script async src="//dn-lbstatics.qbox.me/busuanzi/2.3/busuanzi.pure.mini.js">
</script>


     








</body>
</html>